package com.bjbn.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
@Slf4j
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    @Value("${csv.topic}")
    private String topic;

    public void sendMessage(String key, Object data) {
        try {
            String jsonMessage = objectMapper.writeValueAsString(data);
            log.debug("Sending message to topic {}: {}", topic, jsonMessage);
            
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, jsonMessage);
            
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.debug("Message sent successfully to topic {} partition {} offset {}", 
                            result.getRecordMetadata().topic(),
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                }
                @Override
                public void onFailure(Throwable ex) {
                    log.error("Failed to send message to topic {}: {}", topic, ex.getMessage());
                }
            });
        } catch (JsonProcessingException e) {
            log.error("Error serializing message to JSON: {}", e.getMessage());
        }
    }

    public void sendBatch(String key, Object data) {
        sendMessage(key, data);
    }
} 